package mq.rabbit.demo;

import com.rabbitmq.client.DefaultConsumer;
import mq.rabbit.common.util.RabbitMQUtils;
import org.junit.Test;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * Work queues，也被称为（Task queues），任务模型。
 * 当消息处理比较耗时的时候，可能生产消息的速度会远远大于消息的消费速度。长此以往，消息就会堆积越来越多，无法及时处理。
 * 此时就可以使用work模型：让多个消费者绑定到一个队列，共同消费队列中的消息。队列中的消息一旦消费，就会消失，因此任务是不会被重复执行的。
 * <p>
 * <br />
 * <p>
 * P：生产者：任务的发布者
 * C1：消费者1：领取任务并且完成任务，假设完成速度较慢
 * C2：消费者2：领取任务并且完成任务，假设完成速度较快
 * <p>
 * <br />
 * <p>
 * 默认情况下，RabbitMQ会将每条消息按顺序发送给下一个使用者。 平均每个消费者将获得相同数量的消息。 这种分发消息的方式称为循环。
 * <p>
 * 如果我们想要做到能者多劳，就要改变消息自动确认机制，改为手动确认消息。
 */
public class WordQueue {

    @Test
    public void testProvider() throws IOException {
        // 获取连接
        Connection connection = RabbitMQUtils.getConnection();
        // 获取通道
        Channel channel = connection.createChannel();

        // 通过通道声明队列
        channel.queueDeclare("work", true, false, false, null);

        for (int i = 1; i <= 20; i++) {
            // 生产消息
            String message = "hello work queue-" + i;
            channel.basicPublish("", "work", null, message.getBytes());
        }

        // 关闭
        RabbitMQUtils.closeChanelAndConnection(channel, connection);
    }

    @Test
    public void testConsumer_02() throws IOException, InterruptedException {

        // 获取连接
        Connection connection = RabbitMQUtils.getConnection();

        // 获取通道
        Channel channel = connection.createChannel();

        // 通过通道声明队列
        channel.queueDeclare("work", true, false, false, null);

        // 消费消息
        // 参数1：队列名称
        // 参数2：消息自动确认，true表示消费者自动向MQ确认消息消费
        channel.basicConsume("work", true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                System.out.println("消费者2 --- " + new String(body));
            }
        });

        TimeUnit.MINUTES.sleep(3);
    }

    @Test
    public void testConsumer_01() throws IOException, InterruptedException {

        // 获取连接
        Connection connection = RabbitMQUtils.getConnection();

        // 获取通道
        Channel channel = connection.createChannel();

        // 通过通道声明队列
        channel.queueDeclare("work", true, false, false, null);

        // 消费消息
        // 参数1：队列名称
        // 参数2：消息自动确认，true表示消费者自动向MQ确认消息消费
        channel.basicConsume("work", true, new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("消费者1 --- " + new String(body));
            }
        });

        TimeUnit.MINUTES.sleep(3);

    }
}
